package cn.itcast.spark.rdd

import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class TransformationOp {
  val conf = new SparkConf().setMaster("local[6]").setAppName("transformation_op")
  val sc = new SparkContext(conf)

  @Test
  def mapPartitions(): Unit = {
    // 1、数据生成
    // 2、算子使用
    // 3、获取结果
    sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2)
      .mapPartitions(iter => {
        iter.foreach(item => println(item))
        iter
      })
      .collect()
  }

  @Test
  def mapPartitions2(): Unit = {
    // 1、数据生成
    // 2、算子使用
    // 3、获取结果
    sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2)
      .mapPartitions(iter => {
        // 遍历iter其中每一条数据进行转换，转换完成以后，返回这个iter
        // iter是scala中的集合类型
        iter.map(item => item * 10)
      })
      .collect()
      .foreach(item => println(item))
  }

  @Test
  def mapPartitionsWithIndex(): Unit = {
    // 1、数据生成
    // 2、算子使用
    // 3、获取结果
    sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2)
      .mapPartitionsWithIndex((index, iter) => {
        println("index:" + index)
        iter.foreach(item => println(item))
        iter
      })
      .collect()
  }

  @Test
  def filter(): Unit = {
    sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
      .filter(item => item % 2  == 0)
      .collect()
      .foreach(item => println(item))
  }

  @Test
  def sample(): Unit = {
    sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
      .sample(true, 0.6, 20)  // 第一个参数表示是否有放回抽样
      .collect()
      .foreach(item => println(item))
  }

  @Test
  def mapValues(): Unit = {
    sc.parallelize(Seq(("a", 1), ("b", 2),  ("c", 3)))
      .mapValues(item => item * 10)
      .collect()
      .foreach(println(_))
  }

  /**
   * 交集
   */
  @Test
  def intersection(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7, 8))
    rdd1.intersection(rdd2)
      .collect()
      .foreach(println(_))
  }

  /**
   * 并集
   */
  @Test
  def union(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7, 8))
    rdd1.union(rdd2)
      .collect()
      .foreach(println(_))
  }

  /**
   * 差集
   */
  @Test
  def subtract(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7, 8))
    rdd1.subtract(rdd2)
      .collect()
      .foreach(println(_))
  }

  /**
   * groupByKey 运算结果的格式： (K, (value1, value2))
   * reduceByKey 能不能在Map端做Combiner ：1.能不能减少IO
   * groupByKey 在Map端做Combiner没有意思
   */
  @Test
  def groupByKey(): Unit = {
    sc.parallelize(Seq(("a", 1), ("b", 2),  ("c", 3)))
      .groupByKey()
      .collect()
      .foreach(println(_))
  }

  @Test
  def combineByKey(): Unit = {
    // 1、准备集合
    val rdd = sc.parallelize(Seq(
      ("zhangsan", 99.0),
      ("zhangsan", 96.0),
      ("lisi", 97.0),
      ("lisi", 98.0),
      ("zhangsan", 97.0))
    )

    // 2、算子操作
    //    2.1、createCombiner 转换数据
    //    2.2、mergeValue 分区上的聚合
    //    2.3、mergerCombiners 把所有分区上的结果再次聚合，生成最终结果
    val combinerResult = rdd.combineByKey(
      createCombiner = (curr: Double) => (curr, 1),
      mergeValue = (curr: (Double, Int), nextValue: Double) => (curr._1 + nextValue, curr._2 + 1),
      mergeCombiners = (curr: (Double, Int), agg: (Double, Int)) => (curr._1 + agg._1, curr._2 + agg._2)
    )

    // ("zhangsan“, (99+96+97, 3))
    val resultRDD = combinerResult.map(item => (item._1, item._2._1 / item._2._2))

    resultRDD.collect().foreach(println(_))
  }

  @Test
  def foldByKey(): Unit = {
    sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
      .foldByKey(zeroValue = 10)( (curr, agg) => curr + agg )
      .collect()
      .foreach(println(_))
  }

  /**
   * aggregateByKey(zeroValue)(seqOp, combOp)
   * zeroValue: 指定初始值
   * seqOp：作用于每一个元素，根据初始值，进行计算
   * comOp:将seqOp处理过的结果进行聚合
   *
   * aggregateByKey 特别适合针对每个数据要先处理，后聚合
   */
  @Test
  def aggregateByKey(): Unit = {
    val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
    rdd.aggregateByKey(0.8)((zeroValue, item) => item * zeroValue, (curr, agg) => curr + agg)
      .collect()
      .foreach(println(_))
  }

  @Test
  def join(): Unit = {
    val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))
    val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12)))
    rdd1.join(rdd2)
      .collect()
      .foreach(println(_))
  }

  /**
   * sortBy 可以作用于任何类型RDD，sortByKey只有KV类型数据的RDD中才有
   * SortBy 可以按照任何部分来排序，sortByKey只能按照Key排序
   * sortByKey 写法简单，不用编写函数
   */
  @Test
  def sort(): Unit = {
    val rdd1 = sc.parallelize(Seq(2, 4, 1, 6, 1, 8))
    val rdd2 = sc.parallelize(Seq(("a", 1), ("b", 3), ("c", 2)))

    rdd1.sortBy(item => item)

    rdd2.sortBy(item => item._2).collect().foreach(println(_))

//    rdd2.sortByKey().collect().foreach(println(_))

//    rdd2.map(item => (item._2, item._1)).sortByKey().map(item => (item._2, item._1)).collect().foreach(println(_))
  }

  /**
   * repartition 进行重分区的时候，默认是shuffle的
   * coalesce 进行重分区的时候，默认是不shuffle的，coalesce默认不能增大分区数
   */
  @Test
  def partitioning(): Unit = {
    val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5), 2)

    // repartition
//    println(rdd.repartition(5).partitions.size)

    // coalesce
    println(rdd.coalesce(5, shuffle = true).partitions.size)  // shuffle默认为false

  }


}
